package hasthi.common;

import static hasthi.common.constants.HasthiConstants.LOGGER_NAME;

import hasthi.instrument.JavaInstrumentation;
import hasthi.tools.analyze.DataCollector;

import java.util.Collections;
import java.util.Iterator;
import java.util.SortedSet;
import java.util.Timer;
import java.util.TimerTask;
import java.util.TreeSet;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.Future;
import java.util.concurrent.ThreadPoolExecutor;

import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;

/**
 *  this is a wrapper around a execuater service, in addition it makes sure  if a thread get in a deadlock
 *  it is killed eventually
 */
public class SafeScheduler {
    private Log log = LogFactory.getLog(LOGGER_NAME);
    private final ExecutorService executorService;
    private final SortedSet<TaskInfo> runningJobInfo;
    private final JobCancelledCallback jobcallback;
    
    private static LogFile logfile;
    private static DataCollector waitc;
    private static DataCollector runc;

    
    public SafeScheduler(int workerCount,JobCancelledCallback jobCancelledCallback,boolean isCoordinator){
        executorService = Executors.newFixedThreadPool(workerCount);
        runningJobInfo = Collections.synchronizedSortedSet(new TreeSet<TaskInfo>());
        this.jobcallback = jobCancelledCallback;
        //Following thread kill any action that took more than 15minutes to freeup the threads
        int delay = 5000; // delay for 5 sec.
        Timer timer = new Timer(true);
        timer.scheduleAtFixedRate(new TimerTask() {
            @Override
            public void run() {
                synchronized (runningJobInfo) {
                    Iterator<TaskInfo> it =  runningJobInfo.iterator();
                    while(it.hasNext()){
                        TaskInfo taskInfo = it.next();
                        if(taskInfo.jobHandle.isDone() || taskInfo.jobHandle.isCancelled()){
                            it.remove();
                        }else{
                            if(System.currentTimeMillis() - taskInfo.startTime > 15*60*10000){
                                taskInfo.jobHandle.cancel(true);
                                log.info("Job cancelled "+taskInfo.job);
                                jobcallback.jobCancelled(taskInfo.job);
                            }
                        }
                    }
                }
                //Increase the max pool size if there are too many jobs waiting
                if(executorService instanceof ThreadPoolExecutor){
                    ThreadPoolExecutor executor = ((ThreadPoolExecutor)executorService);
                    long pendingtasks = executor.getQueue().size();
                    if(pendingtasks > 50){
                        if(executor.getPoolSize() < 20){
                            executor.setMaximumPoolSize(executor.getPoolSize() + 2);    
                        }
                        log.info("[Scheduler:"+this+"]"+pendingtasks+" pending "+JavaInstrumentation.getMemoryUsage() +" " + executor.getPoolSize());
                    }else{
                        log.debug("[Scheduler:"+this+"]"+pendingtasks+" pending" +" " + executor.getPoolSize());    
                    }
                    //I think we do not need to reduce the max thread count as they will timeout by itself
                }
            }
        
        }, delay, 10000);
        if(logfile == null){
            logfile = LogFile.getLog("action_" 
                    +(isCoordinator?"corrdinator":"manager")
                    +"_action");
            waitc = new DataCollector("waitc",logfile, DataCollector.WINDOW_SIZE);
            runc = new DataCollector("runc",logfile, DataCollector.WINDOW_SIZE);
        }
    }
    
    public void submit(Runnable task){
        TaskInfo taskWrapper = new TaskInfo(System.currentTimeMillis(),task);
        Future future = executorService.submit(taskWrapper);
        taskWrapper.setJobHandle(future);
        synchronized (runningJobInfo) {
            runningJobInfo.add(taskWrapper);
        }
    }
    public class TaskInfo implements Comparable<TaskInfo>, Runnable{
        public Future<String> jobHandle;
        public long startTime;
        public Runnable job;
        public TaskInfo(long startTime,Runnable job) {
            super();
            this.startTime = startTime;
            this.job = job;
        }
        public int compareTo(TaskInfo o) {
            long match = this.startTime - o.startTime; 
            if(match == 0){
                return 0;
            }
            return match > 0?1:-1;
        }
        public void run() {
           long execTime = System.currentTimeMillis();
           try {
                job.run();
            } catch (Throwable e) {
                log.error(e);
                ErrorManager.error(e);
            }
           long endtime = System.currentTimeMillis();
           waitc.addAReading((execTime-startTime)*1000000);
           runc.addAReading((endtime-execTime)*1000000);
        }
        public void setJobHandle(Future<String> jobHandle) {
            this.jobHandle = jobHandle;
        }
        
    }
    
    public static interface JobCancelledCallback{
        public void jobCancelled(Runnable runnable);
    }
    
}
